package com.xyz.wll.es.util;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xyz.wll.es.util.model.*;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.lucene.search.Query;
import org.elasticsearch.action.admin.cluster.repositories.delete.DeleteRepositoryRequestBuilder;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.*;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryShardContext;
import org.elasticsearch.index.reindex.DeleteByQueryRequestBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.KeyManager;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.io.*;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.security.SecureRandom;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.util.*;

class DefaultTrustManager implements X509TrustManager {
    @Override
    public void checkClientTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {

    }

    @Override
    public void checkServerTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {

    }

    @Override
    public X509Certificate[] getAcceptedIssuers() {
        return null;
        //return new X509Certificate[0];
    }
}

public class EsUtil {
    private static final Logger log = LoggerFactory.getLogger(EsUtil.class);

    public static final String MATCH_ALL_QUERY = "{\"query\": {\"match_all\" : { } }, \"size\": 100 }";
    public static final String CAT_INDICES = "/_cat/indices";

    private RestClient restClient;
    private RestHighLevelClient restHighLevelClient;

    private CredentialsProvider credentialsProvider;
    private SSLContext sslContext;

    public static final int bulkRecordNum = 500;

    public EsUtil() {
        try {
            credentialsProvider = new BasicCredentialsProvider();
            sslContext = SSLContext.getInstance("TLS");
            sslContext.init(new KeyManager[0], new TrustManager[]{new DefaultTrustManager()}, new SecureRandom());
        } catch (KeyManagementException e) {
            e.printStackTrace();
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
        }
    }

    public RestClient getRestClient() {
        return restClient;
    }

    public RestHighLevelClient getRestHighLevelClient() {
        return restHighLevelClient;
    }

    public boolean connect(String esHosts) throws IOException {
        return connect(esHosts, "http", null, null);
    }

    public boolean connect(String esHosts, String scheme, String userName, String password) throws IOException {
        String[] hosts = esHosts.split(",");
        HttpHost[] httpHosts = new HttpHost[hosts.length];
        for(int i = 0; i < hosts.length; i++) {
            String host = hosts[i];
            String[] fields = host.split(":");
            int port = 9200;
            if(fields.length == 2) {
                port = Integer.parseInt(fields[1]);
            }
            httpHosts[i] = new HttpHost(fields[0], port, scheme);
        }
        RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);

        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                if("https".equals(scheme.toLowerCase())) {
                    httpAsyncClientBuilder.setSSLContext(sslContext);
                }
                if(userName != null && !userName.isEmpty()) {
                    for (HttpHost httpHost : httpHosts) {
                        credentialsProvider.setCredentials(new AuthScope(httpHost), new UsernamePasswordCredentials(userName, password));
                    }
                    httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
                return httpAsyncClientBuilder;
            }
        });

        restClient = restClientBuilder.build();
        restHighLevelClient = new RestHighLevelClient(restClientBuilder);
        if (restClient != null && restHighLevelClient != null) {
            String test = get("/","");
            JSONObject result = JSONObject.parseObject(test);
            return !"".equals(result.getString("cluster_name"));
        }
        return false;
    }

    public String request(String method, String path, String paramsJson) throws IOException {

        Request request = new Request(method,path);
        request.setJsonEntity(paramsJson);
        Response response = restClient.performRequest(request);
        InputStream inputStream = response.getEntity().getContent();
        byte[] buffer = new byte[(int)(response.getEntity().getContentLength())];
        int totalBytes = 0;
        do {
            int bytes = inputStream.read(buffer);
            totalBytes += bytes;
        }while(totalBytes < buffer.length);
        return new String(buffer,"UTF-8");

    }

    public String get(String path, String paramsJson) throws IOException {
        return request("GET", path, paramsJson);
    }

    public String put(String path, String paramsJson) throws IOException {
        return request("PUT", path, paramsJson);
    }

    public String post(String path, String paramsJson) throws IOException {
        return request("POST", path, paramsJson);
    }

    public JSONObject getIndexMappings(String index) throws IOException {
        String path = "/" + index + "/" + "_mappings";
        String result = get(path, "");
        JSONObject root = JSONObject.parseObject(result);

        //System.out.println(result);
        JSONObject mappings = root.getJSONObject(index).getJSONObject("mappings");
        return mappings;
    }

    public Map<String,String> getMappings(String index) throws IOException {

        Map<String, String> typeMap = new TreeMap<>();

        JSONObject mappings = getIndexMappings(index);

        //System.out.println(result);
        JSONObject properties = mappings.getJSONObject("properties");
        if(properties != null) {
            for (Map.Entry<String, Object> entry : properties.entrySet()) {
                String type = ((JSONObject) (entry.getValue())).getString("type");
                String key = entry.getKey();
                if (!"_doc".equals(key)) {
                    if("text".equals(type)) {
                        JSONObject f = ((JSONObject) (entry.getValue())).getJSONObject("fields");
                        if(f != null && f.getJSONObject("keyword") != null && "keyword".equals(f.getJSONObject("keyword").getString("type"))) {
                            typeMap.put(key, "string");
                        } else {
                            typeMap.put(key, type);
                        }
                    } else {
                        typeMap.put(key, type);
                    }
                }
            }
        }
        return typeMap;

    }

    public boolean setIndexMappings(String index, JSONObject mappings) {
        String path = "/" + index + "/" + "_mappings";
        try {
            String result = put(path, mappings.toJSONString());
            JSONObject resultJson = JSONObject.parseObject(result);
            return resultJson.getBoolean("acknowledged");
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean setMappings(String index, Map<String,String> mappings) throws IOException {

        JSONObject properties = new JSONObject();
        for(String key : mappings.keySet()) {
            String type = mappings.get(key);
            if("string".equals(type)) {
                JSONObject keywordTypeObject = new JSONObject();
                keywordTypeObject.put("type", "keyword");
                keywordTypeObject.put("ignore_above", 256);
                JSONObject fieldsObject = new JSONObject();
                fieldsObject.put("keyword", keywordTypeObject);
                JSONObject stringTypeObject = new JSONObject();
                stringTypeObject.put("type","text");
                stringTypeObject.put("fields", fieldsObject);
                properties.put(key, stringTypeObject);
            } else {
                JSONObject typeObject = new JSONObject();
                typeObject.put("type", type);
                properties.put(key, typeObject);
            }
        }
        JSONObject mappingsObject = new JSONObject();
        mappingsObject.put("properties", properties);

        return setIndexMappings(index, mappingsObject);
    }

    public JSONObject getSettings(String index) throws IOException {
        String path = "/" + index + "/_settings";
        String result = get(path, "");
        JSONObject jsonObject = JSONObject.parseObject(result);
        return jsonObject.getJSONObject(index).getJSONObject("settings");
    }

    private void filterSettings(JSONObject settings) {
        filterSettings(settings, false);
    }

    private void filterSettings(JSONObject settings, boolean createFlag) {
        JSONObject indexNode = settings.getJSONObject("index");
        indexNode.remove("provided_name");
        indexNode.remove("uuid");
        indexNode.remove("version");
        indexNode.remove("creation_date");
        if(!createFlag) {
            indexNode.remove("number_of_shards");
        }
    }

    public boolean setSettings(String index, JSONObject settings) throws IOException {
        String path = "/" + index + "/_settings";
        filterSettings(settings);
        String result = put(path, settings.toJSONString());
        JSONObject resultJson = JSONObject.parseObject(result);
        return resultJson.getBoolean("acknowledged");
    }

    public String search(String index, String queryJson) throws IOException {
        return get("/" + index + "/_search", queryJson);
    }

    public String search(String index) throws IOException {
        return search(index, MATCH_ALL_QUERY);
    }

    public Map<String, IndexDetail> getIndices() throws IOException {

        TreeMap<String,IndexDetail> indicesDetail = null;
        String result = get(CAT_INDICES,null);
        if(result != null) {
            String[] lines = result.split("\n");
            indicesDetail = new TreeMap<>();
            for(String line : lines) {
                String[] fields = line.split("\\s+");
                if (fields.length >= 10) {
                    IndexDetail indexDetail = new IndexDetail();
                    indexDetail.setHealth(fields[0]);
                    indexDetail.setStatus(fields[1]);
                    indexDetail.setIndex(fields[2]);
                    indexDetail.setUuid(fields[3]);
                    indexDetail.setPri(Integer.valueOf(fields[4]));
                    indexDetail.setRep(Integer.valueOf(fields[5]));
                    indexDetail.setDocsCount(Long.valueOf(fields[6]));
                    indexDetail.setDocsDeleted(Long.valueOf(fields[7]));
                    indexDetail.setStoreSize(fields[8]);
                    indexDetail.setPriStoreSize(fields[9]);
                    indicesDetail.put(indexDetail.getIndex(), indexDetail);
                }
            }
        }
        return indicesDetail;

    }

    public DocRecord getDocument(String index, String id) {
        GetRequest getRequest = new GetRequest(index, id);
        try {
            GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            if(getResponse.isExists()) {
                DocRecord docRecord = new DocRecord();
                docRecord.setIndex(getResponse.getIndex());
                docRecord.setId(getResponse.getId());
                docRecord.setType(getResponse.getType());
                docRecord.setSource(getResponse.getSource());
                return docRecord;
            } else {
                return null;
            }
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    public boolean upsertDocument(DocRecord docRecord) {

        try {
            IndexRequest indexRequest = new IndexRequest(docRecord.getIndex());

            if(docRecord.getId() != null) {
                indexRequest.id(docRecord.getId());
            }
            indexRequest.source(docRecord.getSource());
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest,RequestOptions.DEFAULT);
            docRecord.setId(indexResponse.getId());
            return indexResponse.status().equals(RestStatus.CREATED);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean insertDocument(DocRecord docRecord) {

        if(getDocument(docRecord.getIndex(), docRecord.getId()) != null) {
            return false;
        }

        return upsertDocument(docRecord);
    }

    public boolean updateDocument(DocRecord docRecord) {
        UpdateRequest updateRequest = new UpdateRequest(docRecord.getIndex(),docRecord.getId());
        updateRequest.doc(docRecord.getSource());
        try {
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            return updateResponse.status().equals(RestStatus.OK);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean deleteDocument(String index, String type, String id) {
        DeleteRequest deleteRequest = new DeleteRequest(index, id);
        try {
            DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            return deleteResponse.status().equals(RestStatus.OK);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public long countDocuments(String index) {
        CountRequest countRequest = new CountRequest(index);
        try {
            CountResponse countResponse = restHighLevelClient.count(countRequest, RequestOptions.DEFAULT);
            return countResponse.getCount();
        } catch (IOException e) {
            e.printStackTrace();
            return -1;
        }
    }

    public boolean bulkIndexDocuments(String index, List<DocRecord> docRecords, long startId) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        for(DocRecord docRecord:docRecords) {
            IndexRequest indexRequest = new IndexRequest(index);
            if(docRecord.getId() != null) {
                indexRequest.id(docRecord.getId());
            } /*else {
                indexRequest.id(Long.toString(startId));
            }*/
            indexRequest.source(docRecord.getSource());

            bulkRequest.add(indexRequest);
            startId++;
        }
        try {
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            for(BulkItemResponse bulkItemResponse : bulkResponse.getItems()) {
                if(bulkItemResponse.isFailed()) {
                    return false;
                }
            }
            System.out.println(bulkResponse.status());
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public long importData(String index, String filepath, ProcessFeedback processFeedback) {
        List<DocRecord> docRecords = new ArrayList<>();

        long total = 0;
        if(processFeedback != null) {
            try (FileReader fileReader = new FileReader(filepath)) {
                BufferedReader bufferedReader = new BufferedReader(fileReader);
                while (bufferedReader.readLine()!=null) {
                    total++;
                }
            }  catch (FileNotFoundException e) {
                e.printStackTrace();
                return -1;
            } catch (IOException e) {
                e.printStackTrace();
                return -1;
            }
        }
        total--;
        System.out.println(String.format("total=%d", total));

        try (FileReader fileReader = new FileReader(filepath)) {
            Map<String,String> fieldsMappings = getMappings(index);

            BufferedReader bufferedReader = new BufferedReader(fileReader);

            String head = bufferedReader.readLine();
            String[] columns = Split(head, ',');

            if(!fieldsMappings.keySet().containsAll(Arrays.asList(columns))) {
                if(processFeedback != null) {
                    processFeedback.reportMessage(String.format("表头字段[%s]不是%s的mapping字段[%s]的子集", String.join(",",columns), index, String.join(",",fieldsMappings.keySet())));
                }
                return 0;
            }

            String line = null;
            long count = countDocuments(index);
            long n = 0;
            while ((line = bufferedReader.readLine()) != null) {
                String[] values = Split(line,',');
                if(values.length != columns.length) {
                    System.out.println(String.format("第%d行数据字段数%d不等于表头字段数%d", n, values.length, columns.length));
                    System.out.println(line);
                    System.out.println("此行数据将被跳过");
                    n++;
                    continue;
                }
                DocRecord docRecord = new DocRecord();
                docRecord.setSource(new TreeMap<>());

                for (int i = 0; i < columns.length; i++) {
                    String column = columns[i];
                    String value = values[i].isEmpty() ? null : values[i];

                    if (column.equals("_id")) {
                        docRecord.setId(value);
                    } else {
                        docRecord.getSource().put(column, value);
                    }
                }
                docRecords.add(docRecord);
                n++;
                if (n % bulkRecordNum == 0) {
                    long start_point = System.currentTimeMillis();
                    for(int i =0; i < 3 && !bulkIndexDocuments(index, docRecords, count + n - bulkRecordNum); i++);
                    long end_point = System.currentTimeMillis();
                    docRecords.clear();
                    System.out.println(String.format("%d/%d条记录加载完毕,耗时%d毫秒", n, total, (end_point-start_point)));
                    if(processFeedback != null) {
                        processFeedback.reportPercent((float)n/(total));
                        processFeedback.reportMessage(String.format("%d/%d条记录加载完毕,耗时%d毫秒", n, total, (end_point-start_point)));
                    }
                }
            }
            if(n % bulkRecordNum != 0) {
                long start_point = System.currentTimeMillis();
                for(int i=0; i<3 && !bulkIndexDocuments(index, docRecords, count+n-n%bulkRecordNum); i++);
                long end_point = System.currentTimeMillis();
                docRecords.clear();
                System.out.println(String.format("%d/%d条记录加载完毕,耗时%d毫秒", n, total, (end_point-start_point)));
                if(processFeedback != null) {
                    processFeedback.reportPercent((float)n/(total));
                }
            }
            return n;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
            return -1;
        } catch (IOException e) {
            e.printStackTrace();
            return -1;
        }
    }

    public long exportData(String index, String filepath, ProcessFeedback processFeedback) {
        long n = countDocuments(index);
        long completed = 0;
        if (n > 0) {
            try (FileWriter fileWriter = new FileWriter(filepath)) {
                BufferedWriter bufferedWriter = new BufferedWriter(fileWriter);
                Boolean headLineWrited = false;
                boolean isFirstPage = true;

                Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1));
                RestHighLevelClient client = getRestHighLevelClient();
                SearchResponse searchResponse = null;
                String scrollId = null;

                int pages = 0;
                do {
                    if (isFirstPage) {
                        SearchRequest searchRequest = new SearchRequest();
                        searchRequest.scroll(scroll);
                        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                        searchSourceBuilder.query();
                        searchSourceBuilder.size(1000);
                        searchRequest.source(searchSourceBuilder);
                        searchRequest.indices(index);
                        searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
                        isFirstPage = false;
                    } else {
                        SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
                        searchScrollRequest.scroll(scroll);
                        searchScrollRequest.scrollId(scrollId);

                        searchResponse = client.scroll(searchScrollRequest, RequestOptions.DEFAULT);
                    }

                    if (searchResponse.status().getStatus() == 200) {
                        SearchHit[] searchHits = searchResponse.getHits().getHits();
                        if (searchHits != null) {
                            for (SearchHit searchHit : searchHits) {
                                if (!headLineWrited) {
                                    bufferedWriter.write("_id");
                                    for (String key : searchHit.getSourceAsMap().keySet()) {
                                        bufferedWriter.write(",");
                                        bufferedWriter.write(key);
                                    }
                                    bufferedWriter.write("\n");
                                    headLineWrited = true;
                                }
                                bufferedWriter.write(searchHit.getId());
                                for (String key : searchHit.getSourceAsMap().keySet()) {
                                    Object value = searchHit.getSourceAsMap().get(key);
                                    bufferedWriter.write(",");
                                    bufferedWriter.write(value == null ? "" : value.toString());
                                }
                                bufferedWriter.write("\n");
                            }
                            bufferedWriter.flush();
                            completed += searchHits.length;
                            if(processFeedback != null) {
                                processFeedback.reportPercent((float)completed/n);
                                processFeedback.reportMessage(String.format("导出%d(%d/%d)条数据记录=>%s", searchHits.length, completed, n, index));
                            }
                            System.out.println(String.format("导出%d(%d/%d)条数据记录=>%s", searchHits.length, completed, n, index));
                        }
                    }
                    scrollId = searchResponse.getScrollId();
                } while (scrollId != null && searchResponse.getHits().getHits() != null && searchResponse.getHits().getHits().length > 0);

                if (scrollId != null) {
                    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                    clearScrollRequest.addScrollId(scrollId);
                    ClearScrollResponse clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
                    System.out.println(clearScrollResponse.toString());
                }
                bufferedWriter.close();
                return completed;
            } catch (IOException e1) {
                e1.printStackTrace();
                return -1;
            }
        }
        return n;
    }

    public EsSearchResult search(String index, int from, int size) throws IOException {
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(index);
        searchRequest.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()).from(from).size(size));
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        String result = searchResponse.toString();
        //System.out.println(result);
        EsSearchResult esSearchResult = JSONObject.parseObject(result, EsSearchResult.class);
        return esSearchResult;

    }

    public EsSearchResult search(String index, List<QueryCondition> queryConditions, int from, int size) throws IOException {
        QueryBuilder queryBuilder = getQueryBuilder(queryConditions);

        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(index);
        searchRequest.source(new SearchSourceBuilder().query(queryBuilder).from(from).size(size));
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        String result = searchResponse.toString();
        //System.out.println(result);
        EsSearchResult esSearchResult = JSONObject.parseObject(result, EsSearchResult.class);
        return esSearchResult;

    }

    private QueryBuilder getQueryBuilder(List<QueryCondition> queryConditions) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        for(QueryCondition queryCondition : queryConditions) {
            switch (queryCondition.compare) {
                case QueryCondition.EQ:
                {
                    boolQueryBuilder.must(QueryBuilders.termQuery(queryCondition.field, queryCondition.value));
                    break;
                }
                case QueryCondition.NEQ:
                {
                    boolQueryBuilder.mustNot(QueryBuilders.termQuery(queryCondition.field, queryCondition.value));
                    break;
                }
                case QueryCondition.GT:
                {
                    boolQueryBuilder.must(QueryBuilders.rangeQuery(queryCondition.field).gt(queryCondition.value));
                    break;
                }
                case QueryCondition.LT:
                {
                    boolQueryBuilder.must(QueryBuilders.rangeQuery(queryCondition.field).lt(queryCondition.value));
                    break;
                }
                case QueryCondition.GTE:
                {
                    boolQueryBuilder.must(QueryBuilders.rangeQuery(queryCondition.field).gte(queryCondition.value));
                    break;
                }
                case QueryCondition.LTE:
                {
                    boolQueryBuilder.must(QueryBuilders.rangeQuery(queryCondition.field).lte(queryCondition.value));
                    break;
                }
            }
        }
        return boolQueryBuilder;
    }

    public EsSqlTableRecord sqlQuery(String sql) throws IOException {

        String path = "/_sql";
        JSONObject params = new JSONObject();
        params.put("query", sql);
        String result = get(path, JSONObject.toJSONString(params));
        JSONObject jsonObject = JSONObject.parseObject(result);


        EsSqlTableRecord esSqlTableRecord = new EsSqlTableRecord();

        JSONArray columnsArray = jsonObject.getJSONArray("columns");
        ColumnType[] columnTypes = new ColumnType[columnsArray.size()];
        for(int i=0; i<columnsArray.size(); i++) {
            columnTypes[i] = columnsArray.getObject(i, ColumnType.class);
        }
        esSqlTableRecord.setColumns(columnTypes);

        List<Object[]> allRows = new ArrayList<>();

        JSONArray rowsArray = jsonObject.getJSONArray("rows");
        for(int i=0; i<rowsArray.size(); i++) {
            allRows.add(rowsArray.getJSONArray(i).toArray());
        }

        esSqlTableRecord.setCursor(jsonObject.getString("cursor"));

        while(esSqlTableRecord.getCursor() != null) {
            params.remove("query");
            params.put("cursor",esSqlTableRecord.getCursor());
            result = get(path,JSONObject.toJSONString(params));

            jsonObject = JSONObject.parseObject(result);

            rowsArray = jsonObject.getJSONArray("rows");
            for(int i=0; i<rowsArray.size(); i++) {
                allRows.add(rowsArray.getJSONArray(i).toArray());
            }

            esSqlTableRecord.setCursor(jsonObject.getString("cursor"));
        }

        Object[][] rows = new Object[allRows.size()][];
        for(int i=0; i<allRows.size(); i++) {
            rows[i] = allRows.get(i);
        }
        esSqlTableRecord.setRows(rows);

        log.info("rows length = " + rows.length);
        return esSqlTableRecord;

    }

    public static EsSearchResult convertSearchResult(String result) {
        try {
            return JSONObject.parseObject(result, EsSearchResult.class);
        } catch (Exception e) {
            log.error(e.getMessage());
            return null;
        }
    }

    public boolean deleteIndex(String index) {
        String result = null;
        try {
            result = delete("/" + index);
            JSONObject jsonObject = JSONObject.parseObject(result);
            return Boolean.TRUE.equals(jsonObject.getBoolean("acknowledged"));
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    private String delete(String path) throws IOException {
        return request("DELETE", path,"");
    }

    public boolean createIndex(String index) {
        return createIndex(index, null);
    }

    public boolean createIndex(String index, String mappingSetting) {
        String result = null;
        try {
            result = put("/" + index, mappingSetting);
            JSONObject jsonObject = JSONObject.parseObject(result);
            return Boolean.TRUE.equals(jsonObject.getBoolean("acknowledged"));
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean truncateIndex(String index) {
        //TODO:
        return false;
    }

    public boolean existIndex(String index) {
        try {
            String result = get("/" + index, null);
            JSONObject jsonObject = JSONObject.parseObject(result);
            return (jsonObject.get(index) != null);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean copyCreateIndex(String index, String destIndex) {

        try {
            String indexMappingSetting = get("/" + index,null);
            JSONObject sourceIndex = JSONObject.parseObject(indexMappingSetting);
            JSONObject root = sourceIndex.getJSONObject(index);
            JSONObject setting = root.getJSONObject("settings");
            filterSettings(setting,true);
            System.out.println(root);
            return createIndex(destIndex,root.toJSONString());
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }

    }

    public boolean reindex(String sourceIndex, String destIndex) {
        String path = "/_reindex?wait_for_completion=false";
        JSONObject params = new JSONObject();
        JSONObject source = new JSONObject();
        source.put("index", sourceIndex);
        JSONObject dest = new JSONObject();
        dest.put("index", destIndex);
        params.put("source", source);
        params.put("dest", dest);
        try {
            System.out.println(params);
            String result = post(path, params.toJSONString());
            System.out.println(result);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean copyIndexData(String sourceIndex, String destIndex) {
        return reindex(sourceIndex, destIndex);
    }

    public static String[] Split(String line, char separator) {
        List<String> result = new ArrayList<>();
        char[] buffer = new char[line.length()];
        int buffer_len = 0;

        for(int i=0; i<line.length(); i++) {
            char c = line.charAt(i);
            if(c == separator) {
                result.add(new String(buffer, 0, buffer_len));
                buffer_len = 0;
            } else {
                buffer[buffer_len++] = c;
            }
        }
        result.add(new String(buffer,0,buffer_len));

        String[] fields = new String[result.size()];
        return result.toArray(fields);
    }
}
